package com.sbp.message.message_producer;

import com.alibaba.fastjson.JSON;
import com.dap.utils.AsyncQueue;
import com.sbp.message.api.entity.Message;
import com.sbp.message.api.service.MessageService;
import com.sbp.message.service.impl.MessageStatusAckServiceImpl;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

/**
 * @author wangmin
 * @discription
 * @created 2020-06-08 17:09
 */
public class MessageProducer implements AsyncQueue.AsyncQueueConsumer {
    private static final Logger logger = LoggerFactory.getLogger(MessageProducer.class);

    @Autowired
    private MessageService messageService;
    @Autowired
    private MessageStatusAckServiceImpl messageStatusAckServiceImpl;

    private AsyncQueue queue = new AsyncQueue();

    /**
     * 创建消息并且异步确认和发送
     * **/
    public Message produceMessage(Message message, int acknowledgeDelaySeconds) {
        message = messageService.createMessage(message);
        messageStatusAckServiceImpl.insertAcknowledge(message);

        // 延迟指定秒数之后再确认和发送消息
        queue.enqueueMessage(this, message, System.currentTimeMillis()+acknowledgeDelaySeconds*1000L);

        return message;
    }
    /**
     * 创建消息并且异步确认和发送, 默认延迟10秒
     * **/
    public Message produceMessage(Message message) {
        return produceMessage(message, 10);
    }
    /**
     * 该方法只能在catch中使用, 用来发送消息回滚第三方接口
     * 直接确认和发送消息. 然后插入一条messageRecord, 插入不会影响发送
     * **/
    @Transactional(propagation = Propagation.REQUIRES_NEW)
    public Message produceMessageInCatch(Message message) {
        // 直接确认和发送消息
        message = messageService.createAndSendMessage(message);

        try {
            messageStatusAckServiceImpl.insertAcknowledge(message);
        } catch (Exception e) {
            logger.warn("message="+JSON.toJSONString(message), e);
        }

        return message;
    }

    @Override
    public void consumeAsyncQueueContent(Object o) {
        Message message = (Message) o;
        if (messageStatusAckServiceImpl.acknowledge(message.getId())) {
            messageService.confirmAndSendMessage(message);
        } else {
            logger.info("消息未确认成功, 也许是因为回滚了. message = {}", JSON.toJSONString(message));
        }
    }
}
